package com.z.redis.aof;


import sun.misc.Unsafe;

import javax.naming.event.ObjectChangeListener;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 线程不安全的BlockingQueue，尽量单线程使用，尽量使用 offer和poll两个底层方法
 * Redis 主线程将写命令快速插入 RingBlockingQueue，避免直接写磁盘的 I/O 阻塞
 * @param <E> 泛型
 */
public class RingBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
    // 环形缓冲区
    static final int MAXIMUM_CAPACITY = 1 << 29; //最大容量 (2^29)
    static final int MAXIMUM_SUBAREA = Unsafe.getUnsafe().pageSize(); // 内存页大小(4096)

    Object data[][]; // 二维数组存储元素（分块结构）
    volatile int readIndex = -1; // 读指针 volatile保证可见性
    volatile int writeIndex = -1; // 写指针
    private final AtomicInteger count = new AtomicInteger(); // 元素计数器

    // 双锁机制（分离读/写锁）
    private final ReentrantLock takeLock = new ReentrantLock(); // 取元素锁
    private final Condition notEmpty = takeLock.newCondition(); // 非空条件
    private final ReentrantLock putLock = new ReentrantLock(); // 存元素锁
    private final Condition notFull = putLock.newCondition(); // 非满条件

    // 队列参数
    int capacity; // 实际容量
    int rowOffice, colOffice; // 行/列掩码 （用于位运算）
    int rowSize, bitHigh, subareaSize, maxSize; // 行数、高位位数、分片数、原始容量

    // 通知非空
    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;// takeLock锁 用于控制对队列的取出操作
        takeLock.lock();
        try {
            notEmpty.signal(); // 在队列非空的时候通知等待的线程
        }finally {
            takeLock.unlock(); // 释放锁
        }
    }
    // 通知非满
    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;// putLock锁 用于控制对队列的存入操作
        putLock.lock();
        try {
            notFull.signal(); // 在队列非满的时候通知等待的线程
        }finally {
            putLock.unlock(); // 释放锁
        }
    }

    /**
     * 计算大于等于给定值的最小的2的幂
     */
    static final int tableSizeFor(int cap) {
        int n = cap - 1; // 处理本身是2的幂的情况
        n |= n >>> 1; // 将最高位的1右移1位与原值进行或运算
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }
    /**
     * 返回最低位的1的位
     */
    static final int getIntHigh(int cap) {
        int high = 0;
        while((cap&1) == 0){
            high++;
            cap >>= 1;
        }
        return high;
    }
    /**
     * 计算大于等于给定值的最小的2的幂，但不超过MAXIMUM_SUBAREA
     */
    static final int subareaSizeFor(int cap) {
        int n=cap-1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_SUBAREA) ? MAXIMUM_SUBAREA : n + 1;
    }

    /**
     * 初始化队列
     * @param subareaSize 分片数
     * @param capacity 容量
     */
    public RingBlockingQueue(int subareaSize, int capacity) {
        this(subareaSize, capacity, 1);
    }

    /**
     * 初始化队列
     * @param subareaSize 分片数
     * @param capacity 容量
     * @param concurrency 并发数
     */
    public RingBlockingQueue(int subareaSize, int capacity, int concurrency) {
        if (subareaSize>capacity || subareaSize<0 ){
            // capacity必须大于subareaSize，否则无法存储
            throw new IllegalArgumentException("Illegal initial capacity:subareaSize>capacity||capacity<0||subareaSize<0");
        }
        maxSize = capacity;
        subareaSize = subareaSizeFor(subareaSize);
        capacity = tableSizeFor(capacity);
        rowSize = tableSizeFor(capacity/subareaSize);
        capacity = rowSize * subareaSize;

        this.capacity = capacity;
        this.subareaSize = subareaSize;
        data = new Object[rowSize][subareaSize];
        bitHigh = getIntHigh(subareaSize);
        rowOffice = rowSize - 1;
        colOffice = subareaSize - 1;
    }

    /**
     * 初始化队列
     * @param c
     */
    public RingBlockingQueue(Collection <? extends E> c) {
        // 初始化分片数和容量
        this(8888,88888);

        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            int n=0;
            // 遍历传入的集合元素
            for(E e:c){
                if(e == null){ // 空值检测
                    throw new NullPointerException();
                }
                if(n == capacity){ // 检查队列是否已满
                    throw new IllegalStateException("Queue full");
                }
                // 调用阻塞方法put添加元素
                put(e);
                n++;
            }
            // 元素计数器
            count.set(n);
        }catch (InterruptedException e){ //阻塞方法收到中断请求
            e.printStackTrace();
        }finally {
            putLock.unlock();
        }
    }

    /**
     * 索引回绕调整
     * 当读指针超过队列容量时，读写指针就会回绕到队列起始位置，模拟环形缓冲区的行为
     */
    void refreshIndex(){
        if(readIndex>capacity){ // 检查指针是否超过容量
            putLock.lock();
            try {
                synchronized (this){ // 同步块二次检查
                    if(readIndex>capacity){
                        writeIndex -= capacity;
                        readIndex -= capacity;
                    }
                }
            }finally {
                putLock.unlock();
            }
        }
    }

    /**
     * 入队操作 非阻塞
     * @param o the element to add
     * @return
     */
    @Override
    public boolean offer(Object o) {
        int localWriteIndex = 0;
        synchronized (this){
            if(writeIndex>readIndex+maxSize){ // 检查队列是否已满
                return false;
            }
            count.incrementAndGet(); // 原子递增元素计数器
            localWriteIndex = ++ writeIndex; // 更新写指针
        }
        // 计算行列位置
        int row = (localWriteIndex>>bitHigh)&(rowOffice);
        int col = localWriteIndex&(colOffice);
        if(row == 0&&col == 0){ // 是否到达二维数组起点
            refreshIndex(); // 触发索引回绕
        }
        data[row][col] = o; // 写入数据
        return true;
    }

    /**
     * 出队操作
     * @return
     */
    public E poll(){
        int localReadIndex = 0;
        synchronized(this){
            if(writeIndex <= readIndex ){ // 如果队列为空
                return null;
            }
            localReadIndex = ++readIndex; // 更新读指针
            count.getAndDecrement(); // 原子递减元素计数器
        }
        // 计算行列位置
        int row = (localReadIndex>>bitHigh)&(rowOffice);
        int col = localReadIndex&(colOffice);
        if(row == 0&&col == 0) { // 是否到达二维数组起点 分片边界
            refreshIndex(); // 触发索引回绕
        }
        return (E) data[row][col]; // 返回数据
    }

    /**
     * 遍历指定索引的数据
     * @param index
     * @return
     */
    E ergodic(Integer index){
//        int localReadIndex = 0;
        if(index>writeIndex || index<readIndex){
            return null;
        }
        int row = (index>>bitHigh)&(rowOffice);
        int col = index&(colOffice);
        if(row==0&&col==0){
            refreshIndex();
        }
        return (E) data[row][col];
    }

    /**
     * 获取队列头部元素
     * @return
     */
    @Override
    public E peek(){
        int localReadIndex = 0;
        synchronized (this){
            if(writeIndex <= readIndex ) { // 如果队列为空
                return null;
            }else{
                localReadIndex = readIndex; // 记录当前读指针
            }
        }
        int row = (localReadIndex>>bitHigh)&(rowOffice);
        int col = localReadIndex&(colOffice);
        if(row == 0&&col == 0) {
            refreshIndex();
        }
        return (E) data[row][col];
    }

    /**
     * 阻塞添加元素到队列
     * @param o the element to add
     * @throws InterruptedException
     */
    @Override
    public void put(E o) throws InterruptedException{
        if(o == null){
            throw new NullPointerException();
        }
        // 在所有的put/take等方法中，约定在设置局部变量时将其预设为负数以表示失败，除非成功设置。
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly(); // 获取可中断锁
        try {
            /*
             * 注意，即使在没有锁保护的情况下，count 仍然用于等待条件。
             * 这是因为 count 在此点之后只能减少（所有其他 put 都被锁阻止），
             * 如果 count 发生变化（从容量变为其他值），我们会（或另一个等待的 put）被通知。
             * 类似地，在其他等待条件中使用 count 的所有其他地方也是如此。
             */
            while(count.get() == capacity) { // 如果队列已满
                notFull.await(); // 阻塞等待
            }
            offer(o); // 实际添加元素到队列内部数据结构
            c = count.getAndIncrement(); // 原子递增元素计数器
            if(c+1 < capacity){
                notFull.signal(); // 通知其他线程队列非满
            }
        }finally {
            putLock.unlock(); // 释放锁
        }
        if(c == 0){
            signalNotEmpty(); // 通知其他线程队列非空 唤醒可能阻塞的消费者
        }
    }

    /**
     * 在指定时间将对象添加到队列中
     * @param o the element to add
     * @param timeout how long to wait before giving up, in units of
     *        {@code unit}
     * @param unit a {@code TimeUnit} determining how to interpret the
     *        {@code timeout} parameter
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean offer(E o, long timeout, TimeUnit unit) throws InterruptedException{
        if(o == null){
            throw new NullPointerException();
        }
        long nanos = unit.toNanos(timeout); // 将超时时间转化成纳秒
        int c = -1;
        final ReentrantLock putLock = this.putLock; // 写入锁
        final AtomicInteger count = this.count; // 元素计数器
        putLock.lockInterruptibly(); // 获得可中断锁
        try{
            while(count.get() == capacity) { // 如果队列已满
                if (nanos <= 0) { // 如果超时时间小于等于0
                    return false; // 超时返回失败
                }
                nanos = notFull.awaitNanos(nanos); // 更新剩余时间
            }
            offer(o); // 实际添加元素到队列内部数据结构
            c = count.getAndIncrement(); // 原子递增元素计数器
            if(c+1 < capacity) {
                notFull.signal(); // 通知其他线程队列非满
            }
        }finally {
            putLock.unlock(); // 释放锁
        }
        if(c == 0){
            signalNotEmpty(); // 通知其他线程队列非空 唤醒可能阻塞的消费者
        }
        return true;
    }

    /**
     * 阻塞获取队列头部元素
     * 实现了一个线程安全的阻塞队列take操作
     * 无限期等待 直到队列中有元素
     * @return
     * @throws InterruptedException
     */
    @Override
    public E take() throws InterruptedException{
        E x; // 存储取出的元素
        int c = -1; // 记录操作队列元素数量
        final ReentrantLock takeLock = this.takeLock; // 读取锁 消费者锁
        final AtomicInteger count = this.count; // 元素计数器
        takeLock.lockInterruptibly(); // 获得可中断锁
        try{
            while(count.get() == 0) { // 如果队列为空
                notEmpty.signal();
            }
            x = poll(); // 实际从队列中取出元素
            c = count.getAndDecrement(); // 原子递减元素计数器
            if(c>1){
                notEmpty.signal(); // 通知其他线程队列非空
            }
        }finally {
            takeLock.unlock(); // 释放锁
        }
        if(c == capacity){
            signalNotFull(); // 通知其他线程队列非满 唤醒可能阻塞的生产者
        }
        return x;
    }

    /**
     * 设置了最大等待时间，超时返回null
     * @param timeout how long to wait before giving up, in units of
     *        {@code unit}
     * @param unit a {@code TimeUnit} determining how to interpret the
     *        {@code timeout} parameter
     * @return
     * @throws InterruptedException
     */
    @Override
    public E poll(long timeout, TimeUnit unit) throws InterruptedException{
        E x = null;
        int c = -1;
        long nanos = unit.toNanos(timeout);
        final ReentrantLock takeLock = this.takeLock;
        final AtomicInteger count = this.count;
        takeLock.lockInterruptibly();
        try {
            while(count.get() == 0){
                if(nanos < 0){ // 超时返回null
                    return null;
                }
                nanos = notEmpty.awaitNanos(nanos);
            }
            x = poll();
            c = count.getAndDecrement();
            if(c>1){
                notEmpty.signal();
            }
        }finally {
            takeLock.unlock();
        }
        if(c == capacity){
            signalNotFull();
        }
        return x;
    }

    /**
     * 获得剩余的容量
     * @return
     */
    @Override
    public int remainingCapacity(){
        return capacity - count.get();
    }

    /**
     * 移除指定元素
     * @param o element to be removed from this queue, if present
     * @return
     */
    @Override
    public boolean remove(Object o){
        if(o==null) return false;
        fullyLock();
        try {
            int i = readIndex + 1; // 从第一个未消费的位置开始
            boolean found = false;
            // 遍历环形缓冲区寻找目标元素
            while (i <= writeIndex) {
                int row = (i >> bitHigh) & rowOffice;
                int col = i & colOffice;
                if (o.equals(data[row][col])) {
                    found = true;
                    break;
                }
                i++;
            }
            if (!found) return false;
            // 将目标元素后面的元素依次向前移动一位
            int current = i;
            while (current <= writeIndex) {
                int next = current + 1;
                int nextRow = (next >> bitHigh) & rowOffice;
                int nextCol = next & colOffice;

                int currentRow = (current >> bitHigh) & rowOffice;
                int currentCol = current & colOffice;

                data[currentRow][currentCol] = data[nextRow][nextCol];
                current++;
            }
            writeIndex--;
            count.decrementAndGet();
            //清理最后一个元素引用
            int lastRow = (writeIndex >> bitHigh) & rowOffice;
            int lastCol = writeIndex & colOffice;
            data[lastRow][lastCol] = null;
            return true;
        }finally {
            fullyUnlock();
        }
    }


    private class Itr implements Iterator<E> {
        private final Object[] snapshot;
        private int cursor;

        Itr() {
            fullyLock();
            try {
                snapshot = new Object[count.get()];
                for (int i=0,idx=readIndex+1;idx<=writeIndex;i++,idx++){
                    int row = (idx >> bitHigh) & rowOffice;
                    int col = idx & colOffice;
                    snapshot[i] = data[row][col];
                }
            }finally {
                fullyUnlock();
            }
        }
        public boolean hasNext() {
            return cursor < snapshot.length;
        }

        @SuppressWarnings("unchecked")
        public E next() {
            if(!hasNext()) throw new NoSuchElementException();
            return (E) snapshot[cursor++];
        }
    }
    /**
     * 迭代器实现
     */
    @Override
    public Iterator<E> iterator() {
        return new Itr();
    }

    @Override
    public boolean equals(Object o){
        if(o==this) return true;
        if(!(o instanceof RingBlockingQueue)) return false;

        final BlockingQueue<?> other = (BlockingQueue<?>) o;
        if(this.size()!=other.size()) return false;

        fullyLock();
        try {
            Iterator<?> it = iterator();
            Iterator<?> otherIt = other.iterator();

            while(it.hasNext()&&otherIt.hasNext()){
                if(!Objects.equals(it.next(),otherIt.next())) return false;
            }
            return true;
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public int hashCode(){
        int hashCode = 1;
        for(E e : this){
            hashCode = 31*hashCode+(e==null? 0:e.hashCode());
        }
        return hashCode;
    }

    /**
     * 保留指定队列中的元素
     * @param c collection containing elements to be retained in this collection
     * @return
     */
    @Override
    public boolean retainAll(Collection<?> c){
        Objects.requireNonNull(c); // 显式检查对象是否为null
        fullyLock();
        try{
            boolean modified = false;
            int i = readIndex + 1;
            while(i<=writeIndex){
                int row = (i>>bitHigh)&rowOffice;
                int col = i&colOffice;
                Object e = data[row][col];
                if(!c.contains(e)){
                    // remove
                    remove(e);
                    modified = true;
                    // 元素迁移 重新检查当前位置
                    i--;
                }
                i++;
            }
            return modified;
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public boolean removeAll(Collection<?> c){
        Objects.requireNonNull(c);
        fullyLock();
        try {
            boolean modified = false;
            for (Object e : c) {
                while(remove(e)){ // 循环调用remove操作直到元素不存在
                    modified = true;
                }
            }
            return modified;
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public boolean containsAll(Collection<?> c){
        Objects.requireNonNull(c);
        fullyLock();
        try {
            for(Object e:c){
                if(!contains(e)) return false;
            }
            return true;
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public int size() {
        return count.get();
    }

    @Override
    public boolean isEmpty(){
        return count.get()==0;
    }

    @Override
    public boolean contains(Object o){
        if(o == null) return false;
        fullyLock();
        try {
            for(int i=readIndex;readIndex>=i||i<=writeIndex;i++){
                if(o.equals(ergodic(i))){
                    return true;
                }
            }
            return false;
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public Object[] toArray(){
        fullyLock();
        try {
            int size = count.get();
            Object[] a =new Object[size];
            int k=0;
            for (int index = readIndex; readIndex>=index||index<=writeIndex; index++) {
                a[k++] = ergodic(index);
            }
            return a;
        }finally {
            fullyUnlock();
        }
    }

    public <E> E[] toArray(E[] a){
        fullyLock();
        try {
            int size = count.get();
            if(a.length < size){
                a = (E[])java.lang.reflect.Array.newInstance(a.getClass().getComponentType(),size);
            }

            int k=0;
            for(int index = readIndex; readIndex>=index||index<=writeIndex; index++){
                a[k++] = (E)ergodic(index);
            }
            if(a.length > k){
                a[k] = null;
            }
            return a;
        }finally {
            fullyUnlock();
        }
    }

    /**
     * drainTo()从指定的队列中移除所有可用的元素，并将它们添加到给定的集合中。
     * @param c the collection to transfer elements into
     * @param maxElements the maximum number of elements to transfer
     * @return
     */
    @Override
    public int drainTo(Collection<? super E> c, int maxElements){
        Objects.requireNonNull(c);
        if(c == this) throw new IllegalArgumentException();

        fullyLock();
        try {
            int n = Math.min(maxElements,count.get()); //剩余元素和请求数取最小值

            for(int i=0;i<n;i++){
                int idx=readIndex+i;
                int row=(idx>>bitHigh)&rowOffice;
                int col=(idx&colOffice);
                c.add((E)data[row][col]);
                data[row][col]=null;
                readIndex++;
                count.decrementAndGet();
            }
            if(n>0){
                signalNotFull();
            }
            return n; // 返回实际转移量
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public int drainTo(Collection<? super E> c){
        return drainTo(c,Integer.MAX_VALUE);
    }

    /**
     * 双重锁
     */
    void fullyLock(){
        putLock.lock();
        takeLock.lock();
    }

    void fullyUnlock(){
        putLock.unlock();
        takeLock.unlock();
    }

    @Override
    public boolean add(E e){
        if(offer(e)){
            return true;
        }else{
            throw new IllegalStateException("Queue full");
        }
    }

    @Override
    public E remove(){
        E x=poll();
        if(x!=null){
            return x;
        }else{
            throw new NoSuchElementException();
        }
    }

    @Override
    public E element(){
        E x = peek();
        if(x != null){
            return x;
        }else{
            throw new NoSuchElementException();
        }
    }

    /**
     * 移除队列中的所有元素
     */
    @Override
    public void clear(){
        fullyLock();
        try {
            readIndex = -1;
            writeIndex = -1;
            count.set(0);
            // 显式清理内存引用
            for(int i=0;i<rowOffice;i++){
                Arrays.fill(data[i],null);
            }
            notFull.signal();
        }finally {
            fullyUnlock();
        }
    }

    @Override
    public boolean addAll(Collection<? extends E> c){
        if(c == null) throw new NullPointerException();
        if(c == this) throw new IllegalArgumentException();
        boolean modified = false;
        for(E e:c){
            if(add(e)){
                modified = true;
            }
        }
        return modified;
    }
}
